media_pp\elements\sink\muxer/
hls_muxer.rs

1use std::{
2    borrow::Cow,
3    ffi::CString,
4    path::{Path, PathBuf},
5    ptr,
6    sync::{Arc, Mutex},
7    time::Duration,
8};
9
10use crate::pp_log::{PpLog, pp_error};
11use ffmpeg_next as ffmpeg;
12use thiserror::Error as ThisError;
13
14use crate::{
15    buffer::MediaBuffer,
16    control::ControlMsg,
17    element::{Element, ElementType, Sink, element_pp_log},
18    error::Result,
19};
20
21/// How the media playlist grows and which completed segments remain
22/// referenced by it.
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum HlsMode {
25    /// A sliding live playlist. `window_size` is the maximum number of
26    /// entries kept in the manifest; `delete_old_segments` also removes
27    /// segment files after they fall outside that window.
28    Live {
29        window_size: usize,
30        delete_old_segments: bool,
31    },
32    /// An append-only event playlist. Every segment remains listed and the
33    /// playlist receives `#EXT-X-ENDLIST` when every track finishes.
34    Event,
35    /// A complete video-on-demand playlist. Every segment remains listed
36    /// and the playlist is finalized when every track finishes.
37    Vod,
38}
39
40/// Container used for each HLS media segment.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum HlsSegmentFormat {
43    /// MPEG transport stream segments, conventionally named `*.ts`.
44    MpegTs,
45    /// Fragmented MP4 segments, conventionally named `*.m4s`, plus the
46    /// initialization file selected by [`HlsOptions::init_filename`].
47    Fmp4,
48}
49
50/// Construction-time options for [`HlsMuxer`].
51#[derive(Debug, Clone)]
52pub struct HlsOptions {
53    /// Media playlist written by FFmpeg, normally ending in `.m3u8`.
54    pub playlist_path: PathBuf,
55    /// `printf`-style media segment path containing one integer conversion,
56    /// for example `output/segment_%05d.m4s`.
57    pub segment_pattern: PathBuf,
58    /// Target segment duration. FFmpeg cuts on the next video keyframe, so
59    /// the actual duration can be longer when keyframes are sparse.
60    pub segment_duration: Duration,
61    pub mode: HlsMode,
62    pub segment_format: HlsSegmentFormat,
63    /// fMP4 initialization filename written beside the playlist and used
64    /// in `#EXT-X-MAP`. Ignored for [`HlsSegmentFormat::MpegTs`].
65    pub init_filename: String,
66    /// Optional URI prefix written before segment references in the media
67    /// playlist, for example `https://cdn.example.com/live/`.
68    pub base_url: Option<String>,
69}
70
71impl HlsOptions {
72    /// Creates a live fMP4 configuration with two-second segments, a
73    /// six-segment sliding window, and automatic deletion of old segments.
74    pub fn new(playlist_path: impl Into<PathBuf>, segment_pattern: impl Into<PathBuf>) -> Self {
75        Self {
76            playlist_path: playlist_path.into(),
77            segment_pattern: segment_pattern.into(),
78            segment_duration: Duration::from_secs(2),
79            mode: HlsMode::Live {
80                window_size: 6,
81                delete_old_segments: true,
82            },
83            segment_format: HlsSegmentFormat::Fmp4,
84            init_filename: "init.mp4".into(),
85            base_url: None,
86        }
87    }
88
89    fn validate(&self) -> std::result::Result<(), HlsMuxerError> {
90        if self.segment_duration.is_zero() {
91            return Err(HlsMuxerError::ZeroSegmentDuration);
92        }
93        if let HlsMode::Live { window_size, .. } = self.mode
94            && (window_size == 0 || window_size > i32::MAX as usize)
95        {
96            return Err(HlsMuxerError::InvalidWindowSize(window_size));
97        }
98        if self.segment_format == HlsSegmentFormat::Fmp4 && self.init_filename.is_empty() {
99            return Err(HlsMuxerError::EmptyInitFilename);
100        }
101
102        let pattern = path_as_utf8(&self.segment_pattern, "segment_pattern")?;
103        if !has_integer_conversion(pattern) {
104            return Err(HlsMuxerError::MissingSegmentIndex(
105                self.segment_pattern.clone(),
106            ));
107        }
108        path_as_utf8(&self.playlist_path, "playlist_path")?;
109        reject_nul(pattern, "segment_pattern")?;
110        reject_nul(&self.init_filename, "init_filename")?;
111        if let Some(base_url) = &self.base_url {
112            reject_nul(base_url, "base_url")?;
113        }
114        Ok(())
115    }
116
117    fn header_options(&self) -> std::result::Result<ffmpeg::Dictionary<'static>, HlsMuxerError> {
118        let mut options = ffmpeg::Dictionary::new();
119        options.set("hls_time", &self.segment_duration.as_secs_f64().to_string());
120        let segment_pattern = path_for_ffmpeg(&self.segment_pattern, "segment_pattern")?;
121        options.set("hls_segment_filename", &segment_pattern);
122
123        let mut flags = vec!["temp_file", "independent_segments"];
124        match self.mode {
125            HlsMode::Live {
126                window_size,
127                delete_old_segments,
128            } => {
129                options.set("hls_list_size", &window_size.to_string());
130                if delete_old_segments {
131                    flags.push("delete_segments");
132                }
133            }
134            HlsMode::Event => {
135                options.set("hls_playlist_type", "event");
136                options.set("hls_list_size", "0");
137            }
138            HlsMode::Vod => {
139                options.set("hls_playlist_type", "vod");
140                options.set("hls_list_size", "0");
141            }
142        }
143
144        match self.segment_format {
145            HlsSegmentFormat::MpegTs => options.set("hls_segment_type", "mpegts"),
146            HlsSegmentFormat::Fmp4 => {
147                options.set("hls_segment_type", "fmp4");
148                options.set("hls_fmp4_init_filename", &self.init_filename);
149            }
150        }
151        options.set("hls_flags", &flags.join("+"));
152        if let Some(base_url) = &self.base_url {
153            options.set("hls_base_url", base_url);
154        }
155        Ok(options)
156    }
157}
158
159fn path_as_utf8<'a>(
160    path: &'a Path,
161    field: &'static str,
162) -> std::result::Result<&'a str, HlsMuxerError> {
163    path.to_str().ok_or_else(|| HlsMuxerError::NonUtf8Path {
164        field,
165        path: path.to_path_buf(),
166    })
167}
168
169/// FFmpeg's HLS path splitting recognizes `/` on Windows. Native `\`
170/// separators otherwise make a relative fMP4 init filename land in the
171/// process working directory instead of beside the playlist.
172fn path_for_ffmpeg<'a>(
173    path: &'a Path,
174    field: &'static str,
175) -> std::result::Result<Cow<'a, str>, HlsMuxerError> {
176    let path = path_as_utf8(path, field)?;
177    #[cfg(windows)]
178    {
179        Ok(Cow::Owned(path.replace('\\', "/")))
180    }
181    #[cfg(not(windows))]
182    {
183        Ok(Cow::Borrowed(path))
184    }
185}
186
187fn reject_nul(value: &str, field: &'static str) -> std::result::Result<(), HlsMuxerError> {
188    if value.contains('\0') {
189        Err(HlsMuxerError::EmbeddedNul { field })
190    } else {
191        Ok(())
192    }
193}
194
195/// Allocates the `AVFMT_NOFILE` HLS muxer without opening `playlist_path`
196/// as a normal `AVIOContext`. HLS owns that file and atomically replaces
197/// it; pre-opening it prevents the rename on Windows.
198fn allocate_output(
199    options: &HlsOptions,
200) -> std::result::Result<ffmpeg::format::context::Output, HlsMuxerError> {
201    let path = path_for_ffmpeg(&options.playlist_path, "playlist_path")?;
202    let path = CString::new(path.as_ref()).map_err(|_| HlsMuxerError::EmbeddedNul {
203        field: "playlist_path",
204    })?;
205    let format = CString::new("hls").expect("static HLS format name contains no NUL");
206    let mut context = ptr::null_mut();
207    let result = unsafe {
208        ffmpeg::ffi::avformat_alloc_output_context2(
209            &mut context,
210            ptr::null_mut(),
211            format.as_ptr(),
212            path.as_ptr(),
213        )
214    };
215    if result < 0 {
216        return Err(HlsMuxerError::Ffmpeg(ffmpeg::Error::from(result)));
217    }
218    if context.is_null() {
219        return Err(HlsMuxerError::Ffmpeg(ffmpeg::Error::Unknown));
220    }
221    Ok(unsafe { ffmpeg::format::context::Output::wrap(context) })
222}
223
224/// Accepts `%d`, `%03d`, and the other width variants understood by the
225/// HLS muxer's `printf`-style filename expansion. `%%` is a literal `%`.
226fn has_integer_conversion(pattern: &str) -> bool {
227    let bytes = pattern.as_bytes();
228    let mut index = 0;
229    while index < bytes.len() {
230        if bytes[index] != b'%' {
231            index += 1;
232            continue;
233        }
234        index += 1;
235        if index < bytes.len() && bytes[index] == b'%' {
236            index += 1;
237            continue;
238        }
239        while index < bytes.len() && (bytes[index] == b'0' || bytes[index].is_ascii_digit()) {
240            index += 1;
241        }
242        if index < bytes.len() && bytes[index] == b'd' {
243            return true;
244        }
245    }
246    false
247}
248
249/// Errors specific to [`HlsMuxer`].
250#[derive(Debug, ThisError)]
251pub enum HlsMuxerError {
252    #[error("HlsMuxer stream sinks only accept Packet or Eos buffers, got {0}")]
253    UnsupportedBuffer(&'static str),
254
255    #[error("HLS segment duration must be greater than zero")]
256    ZeroSegmentDuration,
257
258    #[error(
259        "HLS live window size must be between 1 and {max}, got {0}",
260        max = i32::MAX
261    )]
262    InvalidWindowSize(usize),
263
264    #[error("HLS fMP4 init filename must not be empty")]
265    EmptyInitFilename,
266
267    #[error("HLS segment pattern must contain an integer conversion such as %05d: {0:?}")]
268    MissingSegmentIndex(PathBuf),
269
270    #[error("HLS {field} must be valid UTF-8: {path:?}")]
271    NonUtf8Path { field: &'static str, path: PathBuf },
272
273    #[error("HLS {field} must not contain a NUL byte")]
274    EmbeddedNul { field: &'static str },
275
276    #[error("HLS muxer requires at least one stream")]
277    NoStreams,
278
279    #[error("ffmpeg error: {0}")]
280    Ffmpeg(#[from] ffmpeg::Error),
281}
282
283struct PendingStream {
284    name: Arc<str>,
285    input_time_base: ffmpeg::Rational,
286}
287
288/// Builds one HLS media playlist and returns one [`Sink`] per registered
289/// track. FFmpeg owns segment boundary selection, fMP4/MPEG-TS creation,
290/// atomic playlist replacement, live-window trimming, and final
291/// `#EXT-X-ENDLIST` generation.
292///
293/// This deliberately has the same two-phase shape as
294/// [`crate::elements::Mp4Muxer`]: call [`HlsMuxer::add_stream`] for every
295/// track before [`HlsMuxer::open`] writes the header. The returned sinks
296/// share one output lock and finalize the playlist only after every track
297/// reports `Eos` or [`ControlMsg::Stop`].
298pub struct HlsMuxer {
299    output: ffmpeg::format::context::Output,
300    streams: Vec<PendingStream>,
301    options: HlsOptions,
302}
303
304impl HlsMuxer {
305    /// Allocates FFmpeg's HLS output context. Parent directories for the
306    /// playlist, segment pattern, and fMP4 init file must already exist.
307    pub fn create(options: HlsOptions) -> Result<Self> {
308        options.validate()?;
309        let output = allocate_output(&options)?;
310        Ok(Self {
311            output,
312            streams: Vec::new(),
313            options,
314        })
315    }
316
317    /// Registers one encoded packet stream. `time_base` must match the
318    /// timestamps carried by packets arriving at the returned track sink.
319    pub fn add_stream(
320        &mut self,
321        name: impl Into<String>,
322        parameters: ffmpeg::codec::Parameters,
323        time_base: ffmpeg::Rational,
324    ) -> Result<()> {
325        let mut stream = self
326            .output
327            .add_stream(parameters.id())
328            .map_err(HlsMuxerError::from)?;
329        stream.set_time_base(time_base);
330        stream.set_parameters(parameters);
331        self.streams.push(PendingStream {
332            name: name.into().into(),
333            input_time_base: time_base,
334        });
335        Ok(())
336    }
337
338    /// Writes the HLS header and returns one sink per stream, in registration
339    /// order. The playlist is finalized only after every returned sink has
340    /// received `Eos` or [`ControlMsg::Stop`].
341    pub fn open(mut self) -> Result<Vec<Box<dyn Sink>>> {
342        if self.streams.is_empty() {
343            return Err(HlsMuxerError::NoStreams.into());
344        }
345        let options = self.options.header_options()?;
346        let unused = self
347            .output
348            .write_header_with(options)
349            .map_err(HlsMuxerError::from)?;
350        drop(unused);
351        let total = self.streams.len();
352        let shared = Arc::new(HlsMuxerShared {
353            state: Mutex::new(MuxerState {
354                output: self.output,
355                done: 0,
356                finished: false,
357            }),
358            total,
359        });
360        Ok(self
361            .streams
362            .into_iter()
363            .enumerate()
364            .map(|(index, stream)| -> Box<dyn Sink> {
365                Box::new(HlsMuxerStreamSink {
366                    pp_log: element_pp_log(ElementType::HlsMuxer, &stream.name, None),
367                    name: stream.name,
368                    shared: shared.clone(),
369                    stream_index: index,
370                    input_time_base: stream.input_time_base,
371                    done: false,
372                })
373            })
374            .collect())
375    }
376}
377
378struct MuxerState {
379    output: ffmpeg::format::context::Output,
380    done: usize,
381    finished: bool,
382}
383
384struct HlsMuxerShared {
385    state: Mutex<MuxerState>,
386    total: usize,
387}
388
389impl HlsMuxerShared {
390    fn write_packet(
391        &self,
392        stream_index: usize,
393        input_time_base: ffmpeg::Rational,
394        packet: &ffmpeg::Packet,
395    ) -> Result<()> {
396        let mut state = self.state.lock().unwrap();
397        if state.finished {
398            return Ok(());
399        }
400        let mut packet = packet.clone();
401        let output_time_base = state
402            .output
403            .stream(stream_index)
404            .expect("stream was added in HlsMuxer::add_stream")
405            .time_base();
406        packet.rescale_ts(input_time_base, output_time_base);
407        packet.set_stream(stream_index);
408        packet.set_position(-1);
409        packet
410            .write_interleaved(&mut state.output)
411            .map_err(HlsMuxerError::from)?;
412        Ok(())
413    }
414
415    fn finish_track(&self) -> Result<()> {
416        let mut state = self.state.lock().unwrap();
417        state.done += 1;
418        if state.finished || state.done < self.total {
419            return Ok(());
420        }
421        state.finished = true;
422        state.output.write_trailer().map_err(HlsMuxerError::from)?;
423        Ok(())
424    }
425}
426
427/// One registered HLS track's sink. Instances returned by the same
428/// [`HlsMuxer::open`] share their muxer and finalization state.
429pub struct HlsMuxerStreamSink {
430    pp_log: PpLog,
431    name: Arc<str>,
432    shared: Arc<HlsMuxerShared>,
433    stream_index: usize,
434    input_time_base: ffmpeg::Rational,
435    done: bool,
436}
437
438impl HlsMuxerStreamSink {
439    fn finish(&mut self) -> Result<()> {
440        if self.done {
441            return Ok(());
442        }
443        self.done = true;
444        self.shared
445            .finish_track()
446            .inspect_err(|error| pp_error!(self, "write_trailer failed: {error}"))
447    }
448}
449
450impl Element for HlsMuxerStreamSink {
451    fn name(&self) -> Arc<str> {
452        self.name.clone()
453    }
454
455    fn element_type(&self) -> ElementType {
456        ElementType::HlsMuxer
457    }
458
459    fn pp_log(&self) -> &PpLog {
460        &self.pp_log
461    }
462
463    fn pp_log_mut(&mut self) -> &mut PpLog {
464        &mut self.pp_log
465    }
466}
467
468impl Sink for HlsMuxerStreamSink {
469    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
470        match buf {
471            MediaBuffer::Packet(packet) => self
472                .shared
473                .write_packet(self.stream_index, self.input_time_base, &packet)
474                .inspect_err(|error| pp_error!(self, "write_interleaved failed: {error}")),
475            MediaBuffer::Eos => self.finish(),
476            other => Err(HlsMuxerError::UnsupportedBuffer(other.kind()).into()),
477        }
478    }
479
480    fn control(&mut self, msg: ControlMsg) -> Result<()> {
481        if msg == ControlMsg::Stop {
482            self.finish()?;
483        }
484        Ok(())
485    }
486}
487
488#[cfg(test)]
489mod tests {
490    use std::time::{SystemTime, UNIX_EPOCH};
491
492    use super::*;
493    use crate::{
494        element::Source,
495        elements::{AudioCodec, SwAudioEncoder, SwAudioEncoderOptions},
496    };
497
498    fn open_aac_encoder(sample_rate: u32, channels: u16) -> SwAudioEncoder {
499        SwAudioEncoder::new(
500            "encoder",
501            SwAudioEncoderOptions {
502                codec: AudioCodec::Aac,
503                sample_rate,
504                channels,
505                time_base: ffmpeg::Rational::new(1, sample_rate as i32),
506                bit_rate: 64_000,
507            },
508        )
509        .expect("aac encoder must be available")
510    }
511
512    fn silent_frame(
513        sample_rate: u32,
514        channels: u16,
515        samples: usize,
516        pts: i64,
517    ) -> ffmpeg::frame::Audio {
518        let mut frame = ffmpeg::frame::Audio::new(
519            ffmpeg::format::Sample::F32(ffmpeg::format::sample::Type::Packed),
520            samples,
521            ffmpeg::ChannelLayout::default(channels as i32),
522        );
523        frame.set_rate(sample_rate);
524        frame.set_pts(Some(pts));
525        frame.data_mut(0).fill(0);
526        frame
527    }
528
529    fn unique_test_dir(label: &str) -> PathBuf {
530        let nonce = SystemTime::now()
531            .duration_since(UNIX_EPOCH)
532            .unwrap()
533            .as_nanos();
534        std::env::temp_dir().join(format!("media_pp_{label}_{}_{}", std::process::id(), nonce))
535    }
536
537    fn encode_silence(options: HlsOptions, ticks: i64) {
538        let mut encoder = open_aac_encoder(48_000, 1);
539        let mut muxer = HlsMuxer::create(options).expect("HLS muxer must open");
540        muxer
541            .add_stream(
542                "audio",
543                encoder.parameters(),
544                ffmpeg::Rational::new(1, 48_000),
545            )
546            .expect("add_stream must succeed");
547        let mut sinks = muxer.open().expect("HLS header must be written");
548        encoder.src_pads()[0].link(sinks.pop().unwrap());
549
550        for tick in 0..ticks {
551            encoder
552                .consume(MediaBuffer::Audio(Arc::new(silent_frame(
553                    48_000,
554                    1,
555                    960,
556                    tick * 960,
557                ))))
558                .expect("encoding and muxing must succeed");
559        }
560        encoder
561            .consume(MediaBuffer::Eos)
562            .expect("EOS must finalize the HLS playlist");
563    }
564
565    #[test]
566    fn segment_pattern_requires_an_integer_conversion() {
567        assert!(has_integer_conversion("segment_%d.m4s"));
568        assert!(has_integer_conversion("segment_%05d.m4s"));
569        assert!(!has_integer_conversion("segment_%%05d.m4s"));
570        assert!(!has_integer_conversion("segment.m4s"));
571    }
572
573    #[test]
574    fn writes_a_playable_fmp4_vod_playlist() {
575        let dir = unique_test_dir("hls_vod");
576        std::fs::create_dir_all(&dir).unwrap();
577        let playlist_path = dir.join("index.m3u8");
578        let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.m4s"));
579        options.segment_duration = Duration::from_secs(1);
580        options.mode = HlsMode::Vod;
581
582        encode_silence(options, 80);
583
584        let playlist = std::fs::read_to_string(&playlist_path).unwrap();
585        assert!(playlist.starts_with("#EXTM3U"));
586        assert!(playlist.contains("#EXT-X-PLAYLIST-TYPE:VOD"));
587        assert!(playlist.contains("#EXT-X-MAP:URI=\"init.mp4\""));
588        assert!(playlist.contains("#EXT-X-ENDLIST"));
589
590        let segment_uris: Vec<_> = playlist
591            .lines()
592            .filter(|line| !line.is_empty() && !line.starts_with('#'))
593            .collect();
594        assert!(
595            segment_uris.len() >= 2,
596            "expected multiple media segments, got {segment_uris:?}\n{playlist}"
597        );
598        assert!(dir.join("init.mp4").metadata().unwrap().len() > 0);
599        for uri in segment_uris {
600            let path = PathBuf::from(uri);
601            let path = if path.is_absolute() {
602                path
603            } else {
604                dir.join(path)
605            };
606            assert!(
607                path.metadata().unwrap().len() > 0,
608                "empty segment: {path:?}"
609            );
610        }
611
612        let mut input = ffmpeg::format::input(&playlist_path)
613            .expect("FFmpeg must be able to read the generated playlist");
614        assert_eq!(input.streams().count(), 1);
615        let mut packet = ffmpeg::Packet::empty();
616        packet
617            .read(&mut input)
618            .expect("the generated playlist must contain media packets");
619        drop(input);
620        std::fs::remove_dir_all(&dir).unwrap();
621    }
622
623    #[test]
624    fn writes_a_playable_mpegts_vod_playlist() {
625        let dir = unique_test_dir("hls_mpegts");
626        std::fs::create_dir_all(&dir).unwrap();
627        let playlist_path = dir.join("index.m3u8");
628        let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.ts"));
629        options.segment_duration = Duration::from_secs(1);
630        options.mode = HlsMode::Vod;
631        options.segment_format = HlsSegmentFormat::MpegTs;
632
633        encode_silence(options, 80);
634
635        let playlist = std::fs::read_to_string(&playlist_path).unwrap();
636        assert!(playlist.contains("#EXT-X-PLAYLIST-TYPE:VOD"));
637        assert!(!playlist.contains("#EXT-X-MAP"));
638        assert!(playlist.contains("#EXT-X-ENDLIST"));
639        assert!(
640            playlist
641                .lines()
642                .filter(|line| !line.starts_with('#'))
643                .any(|line| line.ends_with(".ts"))
644        );
645
646        let mut input = ffmpeg::format::input(&playlist_path)
647            .expect("FFmpeg must be able to read the generated MPEG-TS playlist");
648        assert_eq!(input.streams().count(), 1);
649        let mut packet = ffmpeg::Packet::empty();
650        packet.read(&mut input).unwrap();
651        drop(input);
652        std::fs::remove_dir_all(&dir).unwrap();
653    }
654
655    #[test]
656    fn live_playlist_keeps_its_window_and_deletes_old_segments() {
657        let dir = unique_test_dir("hls_live");
658        std::fs::create_dir_all(&dir).unwrap();
659        let playlist_path = dir.join("index.m3u8");
660        let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.m4s"));
661        options.segment_duration = Duration::from_secs(1);
662        options.mode = HlsMode::Live {
663            window_size: 2,
664            delete_old_segments: true,
665        };
666
667        encode_silence(options, 200);
668
669        let playlist = std::fs::read_to_string(&playlist_path).unwrap();
670        let segment_uris: Vec<_> = playlist
671            .lines()
672            .filter(|line| !line.is_empty() && !line.starts_with('#'))
673            .collect();
674        assert_eq!(segment_uris.len(), 2, "{playlist}");
675        let media_sequence = playlist
676            .lines()
677            .find_map(|line| line.strip_prefix("#EXT-X-MEDIA-SEQUENCE:"))
678            .unwrap()
679            .parse::<u64>()
680            .unwrap();
681        assert!(media_sequence > 0, "{playlist}");
682        assert!(playlist.contains("#EXT-X-ENDLIST"));
683
684        let media_files = std::fs::read_dir(&dir)
685            .unwrap()
686            .filter_map(|entry| entry.ok())
687            .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "m4s"))
688            .count();
689        assert!(
690            (2..=3).contains(&media_files),
691            "the two listed segments plus at most one deletion-threshold file should remain; \
692             found {media_files}"
693        );
694        assert!(
695            std::fs::read_dir(&dir)
696                .unwrap()
697                .filter_map(|entry| entry.ok())
698                .all(|entry| entry.path().extension().is_none_or(|ext| ext != "tmp")),
699            "atomic temp files must not remain after finalization"
700        );
701        std::fs::remove_dir_all(&dir).unwrap();
702    }
703
704    #[test]
705    fn playlist_finalizes_only_after_every_track_finishes() {
706        let dir = unique_test_dir("hls_tracks");
707        std::fs::create_dir_all(&dir).unwrap();
708        let playlist_path = dir.join("index.m3u8");
709        let mut options = HlsOptions::new(&playlist_path, dir.join("segment_%03d.m4s"));
710        options.segment_duration = Duration::from_secs(1);
711        options.mode = HlsMode::Vod;
712
713        let mut encoder_a = open_aac_encoder(48_000, 1);
714        let mut encoder_b = open_aac_encoder(48_000, 1);
715        let mut muxer = HlsMuxer::create(options).unwrap();
716        muxer
717            .add_stream(
718                "audio-a",
719                encoder_a.parameters(),
720                ffmpeg::Rational::new(1, 48_000),
721            )
722            .unwrap();
723        muxer
724            .add_stream(
725                "audio-b",
726                encoder_b.parameters(),
727                ffmpeg::Rational::new(1, 48_000),
728            )
729            .unwrap();
730        let mut sinks = muxer.open().unwrap();
731        let sink_b = sinks.pop().unwrap();
732        let sink_a = sinks.pop().unwrap();
733        encoder_a.src_pads()[0].link(sink_a);
734        encoder_b.src_pads()[0].link(sink_b);
735
736        for tick in 0..60i64 {
737            encoder_a
738                .consume(MediaBuffer::Audio(Arc::new(silent_frame(
739                    48_000,
740                    1,
741                    960,
742                    tick * 960,
743                ))))
744                .unwrap();
745            encoder_b
746                .consume(MediaBuffer::Audio(Arc::new(silent_frame(
747                    48_000,
748                    1,
749                    960,
750                    tick * 960,
751                ))))
752                .unwrap();
753        }
754
755        encoder_a.consume(MediaBuffer::Eos).unwrap();
756        if let Ok(unfinished) = std::fs::read_to_string(&playlist_path) {
757            assert!(
758                !unfinished.contains("#EXT-X-ENDLIST"),
759                "the first finished track must not finalize the shared playlist"
760            );
761        }
762
763        for tick in 60..80i64 {
764            encoder_b
765                .consume(MediaBuffer::Audio(Arc::new(silent_frame(
766                    48_000,
767                    1,
768                    960,
769                    tick * 960,
770                ))))
771                .unwrap();
772        }
773        encoder_b.consume(MediaBuffer::Eos).unwrap();
774        let finished = std::fs::read_to_string(&playlist_path).unwrap();
775        assert!(finished.contains("#EXT-X-ENDLIST"));
776
777        drop(encoder_a);
778        drop(encoder_b);
779        std::fs::remove_dir_all(&dir).unwrap();
780    }
781}
media_pp\elements\sink\muxer/hls_muxer.rs

media_pp\elements\sink\muxer/
hls_muxer.rs
